package org.example.kafka

import org.apache.zookeeper.Watcher
import org.apache.zookeeper.ZooKeeper
import org.apache.zookeeper.data.Stat
import java.util.concurrent.CountDownLatch

fun main() {
    val connectString = "38.147.171.207:2181" // 改为你的地址

    System.setProperty("java.security.auth.login.config", "zk_client_jaas.conf")
    val latch = CountDownLatch(1)
    val zk = ZooKeeper(connectString, 3000, { event ->
        if (event.state == Watcher.Event.KeeperState.SyncConnected) {
            latch.countDown()
        }
    })

    latch.await()

    // 添加认证信息
//    val username = "admin"
//    val password = "admin123"
//    zk.addAuthInfo("digest", "$username:$password".toByteArray())

    try {
        val children = zk.getChildren("/", false)
        println("Get children for path: / => $children")

//        println("✅ 连接成功，开始遍历所有节点...\n")
//        traverseNode(zk, "/")
    } catch (e: Exception) {
        e.printStackTrace()
    } finally {
        zk.close()
    }
}

/**
 * 递归遍历 ZooKeeper 节点
 */
fun traverseNode(zk: ZooKeeper, path: String) {
    try {
        val stat = Stat()
        val data = zk.getData(path, false, stat)
        val dataType = if (data.isEmpty()) "null" else String(data)
        println("🔍 节点: $path")
        println("   数据: $dataType")
        println("   版本: ${stat.version}, 子节点数: ${stat.numChildren}")

        // 获取子节点
        val children: List<String> = zk.getChildren(path, false)
        if (children.isEmpty()) {
            println("   └── (无子节点)")
        } else {
            for (child in children) {
                val childPath = if (path == "/") "/$child" else "$path/$child"
                traverseNode(zk, childPath) // 递归
            }
        }
        println()

    } catch (e: Exception) {
        println("❌ 访问节点失败 $path: ${e.message}")
    }
}